package com.wfg.bigdata.spark.core.rdd.operator.action

import org.apache.spark.{SparkConf, SparkContext}

object Spark03_RDD_Operator_Action {

    def main(args: Array[String]): Unit = {

        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
        val sc = new SparkContext(sparkConf)

        val rdd = sc.makeRDD(List(1, 2, 3, 4), 2)

        // TODO - 行动算子


        // fold 和reduce 是相同的含义，
        /**
         * fold 方法接受两个参数：
         *
         * 零值（Zero Value）：这是操作开始前的初始值，RDD 中的累积过程将从这个值开始。
         * 二元操作符（Binary Operator）：这是一个函数，它接受两个参数并返回一个值，这个函数将用于将 RDD 中的元素累积到零值上。
         * fold 方法会按照以下步骤工作：
         *
         * 1.将零值与 RDD 的第一个元素应用二元操作符。
         * 2. 将得到的新值与 RDD 的下一个元素应用二元操作符。
         * 3. 重复上述步骤，直到处理完 RDD 中的所有元素。
         */
        val result = rdd.fold(0)(_ + _) // 使用 0 作为零值，使用加法作为二元操作符
        println(result)

        val result1 = rdd.fold(5)(_ + _) // 使用 0 作为零值，使用加法作为二元操作符
        println(result1)


        /**
         * 在 Apache Spark 中，RDD（弹性分布式数据集）的 aggregate 方法是一种转换方法，
         * 它允许你提供一个初始值和一个序列的转换函数，这些函数将应用于 RDD 中的每个元素，以及可选的组合函数，用于合并分区内的结果。
         * aggregate 方法是 fold 方法的推广，它在面对更复杂的数据和计算时提供了更大的灵活性。
         *
         * aggregate 方法的参数：
         * 零值（Zero Value）：这是操作开始前的初始值，类型为 U，aggregate 方法将从这个值开始累积。
         *
         * 序列函数（Seq Function）：这是一个从 RDD 元素类型 T 到中间类型 U 的转换函数，通常表示为 (seqOp, _)。
         * seqOp 是一个二元函数，它将中间类型 U 和 RDD 中的元素 T 结合起来，产生一个新的中间类型 U。
         *
         * 组合函数（Combining Function）：这是一个从中间类型 U 到最终结果类型 V 的转换函数，通常表示为 (_, combOp)。
         * combOp 是一个二元函数，它将两个中间类型 U 的值合并成一个新的中间类型 U。
         */

        //10 + 13 + 17 = 40
        // aggregateByKey : 初始值只会参与分区内计算
        // aggregate : 初始值会参与分区内计算,并且和参与分区间计算
        val result2 = rdd.aggregate(0)(_ + _, _ + _)
        println(result2)


        // 求和 和 平方和
        // 初始化一个元组 (总和, 平方和)
        val zeroValue = (0, 0)

        // 序列函数：计算每个数字的总和和平方和
        val seqOp = (buffer: (Int, Int), element: Int) => (buffer._1 + element, buffer._2 + element * element)

        // 组合函数：合并两个元组的值
        val combOp = (buffer1: (Int, Int), buffer2: (Int, Int)) => (buffer1._1 + buffer2._1, buffer1._2 + buffer2._2)

        val result3 = rdd.aggregate(zeroValue)(seqOp, combOp)

        println(result3) // 输出 (总和, 平方和)，如 (10, 30)



        sc.stop()
    }
}
